package pa4.impl;

import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Timer;
import java.util.TimerTask;

import pa4.interfaces.Node;
import pa4.interfaces.Peer;
import pa4.interfaces.SystemManager;

public class PeerImpl<V> extends UnicastRemoteObject implements Peer<V>, Runnable {

	private ArrayList<Node<V>> nodes = new ArrayList<Node<V>>();
	private int nrLeafNodes = 0;
	//stores the locations of the replications of the peer's nodes
	private HashMap<Node<V>, ArrayList<String>> replicationAddresses = new HashMap<Node<V>, ArrayList<String>>();
	private int numberOfStoredValues = 0;
	private int numberOfReceivedMessages = 0;
	private String name;
	private SystemManager<V> systemManager;
	private HashMap<String, Peer<V>> otherPeersMap = new HashMap<String, Peer<V>>();
	private ArrayList<Peer<V>> otherPeers = new ArrayList<Peer<V>>();
	//timer for the peer's heartbeat signal
	private Timer heartBeatTimer = new Timer();
	private HashMap<String, Timer> heartBeatCountDowns = new HashMap<String, Timer>();
	
	public PeerImpl(String name) throws RemoteException {
		super();
		this.name = name;
	}

	@Override
	public int getNumberOfStoredValues() throws RemoteException {
		return numberOfStoredValues;
	}

	@Override
	public int getNumberOfReceivedMessages() throws RemoteException {
		return numberOfReceivedMessages;
	}

	@Override
	public Node<V> createNode(int id, boolean isLeaf, int level, boolean init) throws RemoteException {
		Node<V> node = new NodeImpl<V>(id, level, this);
		nodes.add(node);
		if(level == 1 && init){
			repairNode(node, "");
		}
		if(level == 6)
			nrLeafNodes++;
		return node;
		/*
		Node<K,V> node = new FTNode<K,V>(id, isLeaf);
		synchronized(nodes){
			nodes.add(node);
		}
		return node;
		*/
	}

	@Override
	public void run() {
		try {
			Registry registry = LocateRegistry.getRegistry("localhost", Registry.REGISTRY_PORT);
			
			SystemManager<V> systemManager = (SystemManager<V>) registry.lookup(SystemManager.SERVICENAME);
			registry.rebind(Peer.SERVICENAME + name, this);
			
//			ArrayList<Peer<V>> otherPeers = systemManager.getRegisteredPeers(this);
//			
//			for(Peer<V> otherPeer : otherPeers){
//				if(!otherPeer.equals(this)){
//					this.otherPeersMap.put(otherPeer.getName(), otherPeer);
//					this.otherPeers.add(otherPeer);
//					otherPeer.connect(this);
//					rescheduleTimerForPeer(otherPeer);
//				}
//			}
//			rescheduleHeartBeatTimer();
			systemManager.join(this);
		} catch (RemoteException e) {
			e.printStackTrace();
		} catch (NotBoundException e) {
			e.printStackTrace();
		}
		
	}
	
	private void rescheduleHeartBeatTimer() {
		heartBeatTimer.schedule(new HeartBeat(this), 5000);
	}

	@Override
	public void repairNodes(String faultyAddress) throws RemoteException {
		for(Node<V> node : nodes){
			repairNode(node, faultyAddress);
		}
	}
	
	public void repairNode(Node<V> node, String faultyAddress) throws RemoteException{
		if(!node.isLeaf()){
			if((node.getLeftChild() == null) ||  
					node.getLeftChildAddress().equalsIgnoreCase(faultyAddress)){
				//System.out.println("Repairing left child with id " + node.getLeftChildId());
				Node<V> leftChildReplication = searchReplication(node.getLeftChildId());
				if(leftChildReplication == null){
					leftChildReplication = recreateNode(node.getLeftChildId(), node.getLevel() + 1);
				}
				node.setLeftChild(leftChildReplication);
				//recursively call the repair function for the new node
				//repairNode(leftChildReplication, faultyAddress);
			}
			if((node.getRightChild() == null) ||  
					node.getRightChildAddress().equalsIgnoreCase(faultyAddress)){
				//System.out.println("Repairing right child with id " + node.getRightChildId());
				Node<V> rightChildReplication = searchReplication(node.getRightChildId());
				if(rightChildReplication == null){
					rightChildReplication = recreateNode(node.getRightChildId(), node.getLevel() + 1);
				}
				node.setRightChild(rightChildReplication);
				//recursively call the repair function for the new node
				//repairNode(rightChildReplication, faultyAddress);
			}
			node.getLeftChild().getPeer().repairNode(node.getLeftChild(), faultyAddress);
			node.getLeftChild().getPeer().repairNode(node.getRightChild(), faultyAddress);
		}
		
	}
	
	private synchronized Node<V> searchReplication(int id) throws RemoteException {
		Iterator iter = otherPeersMap.values().iterator();
		while(iter.hasNext()){
			Peer<V> otherPeer = (Peer<V>) iter.next();
			Node<V> replication = otherPeer.getNodeWithId(id);
			if(replication != null)
				return replication;
		}
		/*
		for(Peer<V> otherPeer : otherPeers){
			Node<V> replication = otherPeer.getNodeWithId(id);
			if(replication != null)
				return replication;
		}*/
		return null;
	}

	public Node<V> recreateNode(int id, int level) throws RemoteException{
		int min;
		Peer<V> newPeer;
		if(level == 6){
			min = getNumberOfLeafNodes();
			newPeer = this;
		}
		
		else{
			min = Integer.MAX_VALUE;
			newPeer = null;
		}
			
		Iterator iter = otherPeersMap.values().iterator();
		while(iter.hasNext()){
			Peer<V> otherPeer = (Peer<V>) iter.next();
			if(level < 6){
				if(min > otherPeer.getNumberOfNodes()){
					newPeer = otherPeer;
					min = otherPeer.getNumberOfNodes();
				}
			}
			else{
				if(min > otherPeer.getNumberOfLeafNodes()){
					newPeer = otherPeer;
					min = otherPeer.getNumberOfLeafNodes();
				}
			}
				
		}
		/*
		for(Peer<V> otherPeer : otherPeers){
			if(min <= otherPeer.getNumberOfNodes())
				newPeer = otherPeer;
		}*/
		if(newPeer != null)
			return newPeer.createNode(id, false, level, true);
		
		//TODO: if a leaf node is recreated, create 3 replicas
		else return null;
	}

	@Override
	public synchronized void messageReceived() throws RemoteException {
		numberOfReceivedMessages++;		
	}

	@Override
	public synchronized void valueStored() throws RemoteException {
		numberOfStoredValues++;		
	}

	@Override
	public synchronized void valueRemoved() throws RemoteException {
		numberOfStoredValues--;
	}

	@Override
	public String getName() throws RemoteException {
		return name;
	}

	@Override
	public Node<V> replicateNode(int id, boolean isLeaf) throws RemoteException {
		// TODO forward copy to system manager who will assign the copy to a new node
		return systemManager.replicateNode(id, isLeaf, this);
	}

	@Override
	public synchronized int getNumberOfNodes() throws RemoteException {
		return nodes.size();
	}

	@Override
	public synchronized void connect(Peer<V> peer) throws RemoteException {
		otherPeersMap.put(peer.getName(), peer);
		//otherPeers.add(peer);
		rescheduleTimerForPeer(peer);
	}

	@Override
	public Node<V> getNodeWithId(int id) throws RemoteException {
		for(Node<V> node : nodes){
			if(node.getId() == id)
				return node;
		}
		return null;
	}
	
	class HeartBeat extends TimerTask{
		private PeerImpl<V> context;
		
		public HeartBeat(PeerImpl<V> context){
			this.context = context;
		}
		@Override
		public void run() {
			try {
				context.sendHeartBeats();
			} catch (RemoteException e) {
				e.printStackTrace();
			}
		}
		
	}
	
	class HeartBeatTimeOut extends TimerTask{
		private PeerImpl<V> context;
		private String otherPeerName;
		
		public HeartBeatTimeOut(PeerImpl<V> context, String otherPeerName){
			this.context = context;
			this.otherPeerName = otherPeerName;
		}

		@Override
		public void run() {
			context.timeOutFor(otherPeerName);
			
		}
		
	}

	public void sendHeartBeats() throws RemoteException {
		Iterator iter = otherPeersMap.values().iterator();
		while(iter.hasNext()){
			Peer<V> otherPeer = (Peer<V>) iter.next();
			otherPeer.heartBeat(this);
		}
		rescheduleHeartBeatTimer();
		/*
		for(Peer<V> otherPeer : otherPeers){
			otherPeer.heartBeat(this);
		}*/
	}

	public synchronized void timeOutFor(String otherPeerName) {
		System.out.println(name + ": " + otherPeerName + " crashed");
		try {
			repairNodes(otherPeerName);
		} catch (RemoteException e) {
			e.printStackTrace();
		}
	}

	@Override
	public synchronized void heartBeat(Peer<V> otherPeer) throws RemoteException {
		Timer t = heartBeatCountDowns.get(otherPeer.getName());
		if(t != null)
			t.cancel();
		/*
		for(Peer<V> peer : otherPeers){
			if(peer.equals(otherPeer)){
				if(heartBeatCountDowns.get(peer) != null)
					heartBeatCountDowns.get(peer).cancel();
			}
		}*/
		rescheduleTimerForPeer(otherPeer);
	}
	
	private synchronized void rescheduleTimerForPeer(Peer<V> otherPeer) throws RemoteException{
		heartBeatCountDowns.put(otherPeer.getName(), new Timer());
		heartBeatCountDowns.get(otherPeer.getName()).schedule(new HeartBeatTimeOut(this, otherPeer.getName()), 10000);
	}

	@Override
	public void pushRegisteredPeerList(ArrayList<Peer<V>> otherPeers)
			throws RemoteException {
		for(Peer<V> otherPeer : otherPeers){
			if(!otherPeer.getName().equalsIgnoreCase(name)){
				this.otherPeersMap.put(otherPeer.getName(), otherPeer);
				this.otherPeers.add(otherPeer);
				otherPeer.connect(this);
				rescheduleTimerForPeer(otherPeer);
			}
		}
		rescheduleHeartBeatTimer();
	}

	@Override
	public synchronized void removeNode(int id) throws RemoteException {
		int level = 0;
		Node<V> removeNode = null;
		for(Node<V> node : nodes){
			if(node.getId() == id){
				removeNode = node;
				level = removeNode.getLevel();
			}
		}
		
		nodes.remove(removeNode);
		if(level == 6)
			nrLeafNodes--;
	}

	@Override
	public synchronized int getNumberOfLeafNodes() throws RemoteException {
		return nrLeafNodes;
	}
	
	//public void balanceNodes

}
